﻿using Analyzor.domain;
using Analyzor.entity;
using Qoollo.Turbo.Collections.Concurrent;
using SOAFramework.Library;
using SqlSugar;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.Drawing;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;

namespace Analyzor
{
    public partial class frmSync : Form
    {
        BackgroundWorker worker = new BackgroundWorker();
        BackgroundWorker pushWorker = new BackgroundWorker();
        EventHandler handler;
        BlockingQueue<CombineData> queue = new BlockingQueue<CombineData>();
        SimpleLogger logger = new SimpleLogger();

        int speed = 0;
        int speedPre1 = 0;
        int speedPre2 = 0;
        int dataCount = 0;
        int tableCount = 0;
        int pageSize = 100000;
        int threadCount = 0;

        bool completed = false;
        static object locker = new object();
        public frmSync()
        {
            InitializeComponent();
            timer1Sec.Tick += Timer1Sec_Tick;
            worker.DoWork += Worker_DoWork;
            //pushWorker.DoWork += PushWorker_DoWork;
        }

        private void PushWorker_DoWork(object sender, DoWorkEventArgs e)
        {
            StartQueue();
        }

        private void Worker_DoWork(object sender, DoWorkEventArgs e)
        {
            CheckTable();
            handler.Invoke(sender, e);
        }

        private void Timer1Sec_Tick(object sender, EventArgs e)
        {
            lblSpeed.Text = ((speed + speedPre1 + speedPre2) / 3).ToString();
            speedPre2 = speedPre1;
            speedPre1 = speed;
            speed = 0;
            pgTable.Maximum = 44;
            pgTable.Value = tableCount;
        }

        private void btnStart_Click(object sender, EventArgs e)
        {
            handler = new EventHandler(Start);
            worker.RunWorkerAsync();
            //pushWorker.RunWorkerAsync();
        }

        void Start(object sender, EventArgs e)
        {
            string tableName = Min5Data.BuildTableName(1);
            DbContext.TargetClient.Queryable<Min5Data>().AS(tableName).ToPageList(1, 1);
            Parallel.For(1, 44, new ParallelOptions { MaxDegreeOfParallelism = 6 }, (i, loopstate) =>
            {
                lock (locker)
                {
                    threadCount++;
                }
                Sync(i);
                lock (locker)
                {
                    threadCount--;
                    tableCount++;
                }
            });
            MessageBox.Show("over");
        }

        void StartSingle(object sender, EventArgs e)
        {
            int index = Convert.ToInt32(txbSharding.Text);
            Sync(index);
        }

        void Sync(int i)
        {
            var src = DbContext.GetClient("src");
            var target = DbContext.GetClient("target");
            try
            {
                string tableName = Min5Data.BuildTableName(i);
                var id = target.Queryable<Min5Data>().AS(tableName).Max(t => t.Id);
                var list = src.Queryable<Min5Data>().AS(tableName).Where(t => t.Id > id).OrderBy(t => t.Id, OrderByType.Asc).ToPageList(1, pageSize);
                while (list.Count > 0)
                {
                    SaveInner(list, target, tableName);
                    //queue.Add(new CombineData
                    //{
                    //    Index = i,
                    //    List = list,
                    //});
                    id = list[list.Count - 1].Id;
                    list = src.Queryable<Min5Data>().AS(tableName).Where(t => t.Id > id).OrderBy(t => t.Id, OrderByType.Asc).ToPageList(1, pageSize);
                }

            }
            catch (Exception ex)
            {
                src.Close();
                target.Close();
                logger.WriteException(ex);
                Thread.Sleep(5000);
                Sync(i);

            }
        }

        void SaveInner(List<Min5Data> list, SqlSugarClient target, string tableName)
        {
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();
            var success = target.Insertable(list).AS(tableName).UseMySql().ExecuteBlueCopy();
            if (success)
            {
                stopwatch.Stop();
                logger.DebugAsync($"耗时:{stopwatch.ElapsedMilliseconds} 写入数据:{list.Count}");
                lock (locker)
                {
                    speed += list.Count;
                }
            }
            else
            {
                throw new Exception("没有成功插入数据");
            }
        }

        void CheckTable()
        {
            var tableList = DbContext.TargetClient.DbMaintenance.GetTableInfoList(false);
            Dictionary<string, DbTableInfo> tableDic = new Dictionary<string, DbTableInfo>();
            foreach (var table in tableList)
            {
                tableDic[table.Name] = table;
            }
            for (int i = 1; i <= 43; i++)
            {
                if (tableDic.ContainsKey(Min5Data.BuildTableName(i)))
                {
                    continue;
                }
                string sql = File.ReadAllText("min5data.sql");
                sql = sql.Replace("{table_name}", Min5Data.BuildTableName(i));
                DbContext.TargetClient.Ado.ExecuteCommand(sql);
            }
        }

        public void StartQueue()
        {
            int maxThread = 5;
            List<Task> tasks = new List<Task>();
            ThreadPool.SetMaxThreads(4, 6);
            while (!completed)
            {
                CombineData data = queue.Take();

                //ThreadPool.QueueUserWorkItem(new WaitCallback(PushData), data);
                //tasks.Add(Task.Factory.StartNew(() =>
                //{
                //    PushData(data);
                //}));
                //if (tasks.Count > 5)
                //{
                //    Task.WaitAny(tasks.ToArray());
                //    tasks = tasks.Where(t => t.Status == TaskStatus.Running).ToList();
                //}
                PushData(data);

            }
            Task.WaitAll(tasks.ToArray());
            MessageBox.Show("数据传输完成");
        }

        public void PushData(Object obj)
        {
            var data = obj as CombineData;
            using (var client = DbContext.GetClient("target"))
            {
                SaveInner(data.List, client, Min5Data.BuildTableName(data.Index));
            }

        }

        private void frmSync_Load(object sender, EventArgs e)
        {
            timer1Sec.Start();
        }

        private void btnSingle_Click(object sender, EventArgs e)
        {
            handler = new EventHandler(StartSingle);
            worker.RunWorkerAsync();
        }

        private void btnCheck_Click(object sender, EventArgs e)
        {
            CheckBySharding();
            MessageBox.Show("over");
        }

        public void CheckBySharding()
        {
            var checkList = DbContext.StockClient.Queryable<entity.Check>().ToList();
            Dictionary<string, entity.Check> dic = new Dictionary<string, entity.Check>();
            checkList.ForEach(t =>
            {
                dic[t.TableName] = t;
            });
            Parallel.For(1, 44, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (i, loopstate) =>
            {
                try
                {
                    var src = DbContext.GetClient("src");
                    var target = DbContext.GetClient("target");
                    string tableName = Min5Data.BuildTableName(Convert.ToInt32(i));

                    if (dic.ContainsKey(tableName))
                    {
                        return;
                    }
                    var srcCount = src.Queryable<Min5Data>().AS(tableName).Count();
                    var targetCount = target.Queryable<Min5Data>().AS(tableName).Count();
                    entity.Check check = new entity.Check
                    {
                        SrcCount = srcCount,
                        TableName = tableName,
                        TargetCount = targetCount,
                    };
                    DbContext.StockClient.Insertable(check).ExecuteCommand();
                    if (!srcCount.Equals(targetCount))
                    {
                        logger.DebugAsync($"tableName:{tableName} src:{srcCount}  target:{targetCount} 差额:{srcCount - targetCount}");
                    }
                } catch (Exception ex)
                {
                    logger.WriteExceptionAsync(ex);
                }

            });
        }

        public void CheckByStock()
        {

            var list = DbContext.StockClient.Queryable<Sharding>().ToList();
            Parallel.For(1, list.Count, new ParallelOptions { MaxDegreeOfParallelism = 1 }, (i, loopstate) =>
            {
                var sharding = list[i];
                var src = DbContext.GetClient("src");
                var target = DbContext.GetClient("target");
                string tableName = Min5Data.BuildTableName(Convert.ToInt32(sharding.ShardingTable));
                var srcCount = src.Queryable<Min5Data>().AS(tableName).Where(t => t.StockId.Equals(sharding.StockCode)).Count();
                var targetCount = target.Queryable<Min5Data>().AS(tableName).Where(t => t.StockId.Equals(sharding.StockCode)).Count();
                if (!srcCount.Equals(targetCount))
                {
                    logger.DebugAsync($"stockCode:{sharding.StockCode}  tableName:{tableName} src:{srcCount}  target:{targetCount} 差额:{srcCount - targetCount}");
                }

            });
        }
    }
}
